package com.czm.art_light.manager.mq.consumer;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.Time;
import co.elastic.clients.elasticsearch.core.BulkRequest;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.core.UpdateRequest;
import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem;
import com.czm.art_light.constant.AmqpConsts;
import com.czm.art_light.constant.ArtErrorCodeEnum;
import com.czm.art_light.constant.CommonConsts;
import com.czm.art_light.constant.EsConsts;
import com.czm.art_light.entity.ArtStatistics;
import com.czm.art_light.entity.ArtWork;
import com.czm.art_light.exception.BusinessException;
import com.czm.art_light.mapper.ArtStatisticsMapper;
import com.czm.art_light.mapper.ArtWorkMapper;
import com.czm.art_light.param.request.ArtWorkESReqDto;
import com.czm.art_light.param.response.ArtWorkEsRespDto;
import com.czm.art_light.service.ArtWorkService;
import com.czm.art_light.util.JsonUtil;
import com.xxl.job.core.biz.model.ReturnT;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * AMQP 消息消费者类
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class AmqpMsgConsumer {

    @Autowired
    private  ElasticsearchClient elasticsearchClient;

    @Autowired
    private ArtStatisticsMapper artStatisticsMapper;

    @Autowired
    private ArtWorkMapper artWorkMapper;

    @Autowired
    private ArtWorkService artWorkService;

    /**
     * 艺术作品 消费失败的消息
     */
    @RabbitListener(queues = AmqpConsts.artWorksMq.DIE_QUEUE)
    public void artWorkUpdateErrorMessageConsumer(String message) {
        log.error("ES中艺术作品信息更新失败的消息：{}", message);
    }

    /**
     * 根据艺术作品Id更新Es艺术作品索引的信息
     *
     * @param message 接受到的消息
     */
    @RabbitListener(queues = AmqpConsts.artWorksMq.ART_WORK_INFO_CHANGE_QUEUE)
    public void artWorkUpdateMessageConsumer(String message) {
        //message 序列化
        ArtWorkESReqDto artWorkESReqDto = JsonUtil.jsonToObj(message, ArtWorkESReqDto.class);
        if (artWorkESReqDto == null) {
            return;
        }
        log.info("ES中艺术作品信息更新开始：{}", artWorkESReqDto);

        try {
            if (artWorkESReqDto.getUpdateAllFields()) {
                updateAllFieldsInES(artWorkESReqDto);
            } else {
                updateStatisticsInES(artWorkESReqDto);
            }
        } catch (IOException e) {
            log.error("更新ES文档失败", e);
            throw new BusinessException(ArtErrorCodeEnum.ES_UPDATE_ARTWORK_ERROR);
        }

        log.info("ES中艺术作品信息更新结束");
    }

    /**
     * 更新ES中的所有字段
     *
     * @param artWorkESReqDto 传入的艺术作品请求DTO
     */
    private void updateAllFieldsInES(ArtWorkESReqDto artWorkESReqDto) throws IOException {
        ArtWork artWork = artWorkMapper.getById(artWorkESReqDto.getWorkId());
        ArtWorkEsRespDto artWorkEsRespDto = artWorkService.buildArtWorkEsRespDto(artWork);

        BulkRequest.Builder br = new BulkRequest.Builder();
        br.operations(op -> op
                .index(idx -> idx
                        .index(EsConsts.ArtWorkIndex.INDEX_NAME)
                        .id(artWorkEsRespDto.getId().toString())
                        .document(artWorkEsRespDto))
        ).timeout(Time.of(t -> t.time(CommonConsts.ELASTICSEARCH_ADD_TIMEOUT)));

        BulkResponse result = elasticsearchClient.bulk(br.build());
        if (result.errors()) {
            log.error("Bulk had errors");
            for (BulkResponseItem item : result.items()) {
                if (item.error() != null) {
                    log.error(item.error().reason());
                }
            }
        }
    }

    /**
     * 更新ES中的统计字段
     *
     * @param artWorkESReqDto 传入的艺术作品请求DTO
     */
    private void updateStatisticsInES(ArtWorkESReqDto artWorkESReqDto) throws IOException {
        ArtStatistics artStatistics = artStatisticsMapper.getByWorkId(artWorkESReqDto.getWorkId());

        Map<String, Object> updatesDoc = getUpdatesDoc(artStatistics);

        // 更新 Elasticsearch 文档
        elasticsearchClient.update(u -> u
                .index(EsConsts.ArtWorkIndex.INDEX_NAME)
                .id(String.valueOf(artWorkESReqDto.getWorkId()))
                .doc(updatesDoc), ArtWorkEsRespDto.class);
    }

    @NotNull
    private static Map<String, Object> getUpdatesDoc(ArtStatistics artStatistics) {
        Map<String, Object> updatesDoc = new HashMap<>();
        // 更新点击量字段
        updatesDoc.put(EsConsts.ArtWorkIndex.FIELD_CLICK_COUNT, artStatistics.getViewCount());
        // 更新下载量字段
        updatesDoc.put(EsConsts.ArtWorkIndex.FIELD_DOWNLOAD_COUNT, artStatistics.getDownloadCount());
        // 更新分享量字段
        updatesDoc.put(EsConsts.ArtWorkIndex.FIELD_SHARE_COUNT, artStatistics.getShareCount());
        // 更新喜欢量字段
        updatesDoc.put(EsConsts.ArtWorkIndex.FIELD_LIKE_COUNT, artStatistics.getLikeCount());
        return updatesDoc;
    }
}
